package cn.lyjuan.aliyun.mq.conf.java.conf.consumer;

import cn.lyjuan.aliyun.mq.conf.java.conf.AMQUser;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * 消费者适配器
 * Created by chad on 2016/11/9.
 */
public class AMQConsumer extends AMQUser implements Consumer
{
    /**
     * 订阅属性
     */
    protected Subscription    sub;

    /**
     * 阿里云消费者代理
     */
    protected ConsumerBean consumer;

    /**
     * 消息监听处理
     */
    protected MessageListener messageListener;

    public Map<Subscription, MessageListener> getSubscriptionTable() {
        return consumer.getSubscriptionTable();
    }

    public void setSubscriptionTable(Map<Subscription, MessageListener> subscriptionTable) {
        consumer.setSubscriptionTable(subscriptionTable);
    }

    @Override
    public void start()
    {
        prepare();// 前置任务

        this.sub = new Subscription();
        sub.setTopic(mqUserCst.getTopicId());// 主题
        sub.setExpression(mqUserCst.getMsgTag());// 消息二级标签

        Properties pro = new Properties();
        pro.setProperty(PropertyKeyConst.ConsumerId, mqUserCst.getUserId());
        pro.setProperty(PropertyKeyConst.AccessKey, mqAuthCst.getAccessKey());
        pro.setProperty(PropertyKeyConst.SecretKey, mqAuthCst.getSecretKey());
        pro.setProperty(PropertyKeyConst.ConsumeThreadNums, Integer.toString(mqUserCst.getConsumeThreadNums()));

        this.consumer = new ConsumerBean();
        consumer.setProperties(pro);

        Map<Subscription, MessageListener> table = new HashMap<>();
        table.put(sub, messageListener);
        this.setSubscriptionTable(table);

        consumer.start();
    }

    public MessageListener getMessageListener()
    {
        return messageListener;
    }

    public void setMessageListener(MessageListener messageListener)
    {
        this.messageListener = messageListener;
    }

    @Override
    public void shutdown()
    {
        consumer.shutdown();
    }

    @Override
    public void subscribe(String s, String s1, MessageListener messageListener)
    {
        consumer.subscribe(s, s1, messageListener);
    }

    @Override
    public void unsubscribe(String s)
    {
        consumer.unsubscribe(s);
    }

    @Override
    public boolean isStarted()
    {
        return consumer.isStarted();
    }

    @Override
    public boolean isClosed()
    {
        return consumer.isClosed();
    }
}
